package com.swsc.ai.tasks;

import com.swsc.ai.config.SparkSqlConf;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;

import java.sql.SQLException;

/**
 * @describe: 以客户为主体静态特征计算
 * @author: DuanCXin
 * @created: 2023-10-26 15:15
 */
public class CustomerEntityTask extends WideTableTempTask {
    @Override
    public void createView(SparkSession session, String... str) throws SQLException {
        long startTime = System.currentTimeMillis();
        String columns_base = "client_id,cust_type, normal_status, marg_status, cust_channel, risk_level, edu_code, " +
                "career_code, age, gender, branch_no, income_trade_status, buyback_trade_status, " +
                "income_active_falg, buyback_active_falg, is_prod_cash, is_qrpagent, emp_name";
        createRealView(session, "dws_cust_portrait_cust_base_info", columns_base, str[0]);
        String columns_ast = "client_id, total_ast, net_ast, nor_total_ast, marg_net_ast, cap_bal, nor_cap_bal, marg_cap_bal, " +
                "total_ast_peak, marg_net_ast_peak, mfin_val, stock_position_rate_last_y, avg_marg_cap_bal_m, avg_marg_cap_bal_y, " +
                "nor_ast_branch_level, nor_ast_company_level, marg_ast_branch_level ,marg_ast_company_level, total_ast_branch_level, total_ast_company_level, " +
                "avg_marg_net_ast_m, avg_nor_cap_bal_m, avg_nor_cap_bal_y, avg_total_ast_y, total_ast_label, marg_net_ast_label, " +
                "marg_cap_bal_label, stock_position_label, stock_hold_label, sec_mkt_val";
        createRealView(session, "dws_cust_portrait_ast_info", columns_ast, str[0]);
        String columns_ablity = "client_id, profit_rate_y_label, loss_rate_y_label, buy_product_risk_label";
        createRealView(session, "dws_cust_portrait_cust_behavior_ablity_info", columns_ablity, str[0]);
        String columns_entrust = "client_id, all_entrst_num, all_entrst_amt";
        createRealView(session, "dws_cust_portrait_cust_entrust_terminal", columns_entrust, str[0]);
        String columns_cust = "client_id, no_trd_days, no_login_days, buy_habit_label, activity_label, turnover_label, marg_sec_perfer_label, marg_ratation_label, marg_debit_change_label, total_cms_y, total_match_amt_y, asset_outflow_rate, turnover_y, activity_y, marg_ratation_m";
        createRealView(session, "dws_cust_portrait_cust_trd_info", columns_cust, str[0]);
        long endTime = System.currentTimeMillis();
        long totalTime = endTime - startTime;
        System.out.println("创建独立视图成功！运行时间： " + totalTime + " 毫秒");
    }

    public void createRealView(SparkSession session, String tableName, String selectColumns, String dt) throws SQLException {
        String condition = "PARTITION (`"+dt+"`)";
        Dataset<Row> df = SparkSqlConf.getDataByPartition(session, tableName, "", selectColumns, condition);
        df.createOrReplaceTempView(tableName);
    }

    @Override
    public void calculateResult(SparkSession session, String dt, String fileName) {
        Dataset<Row> sqlDF = session.sql("select m1.normal_status, m1.cust_type, m1.marg_status, m1.cust_channel, m1.risk_level,\n" +
                "       m1.edu_code, m1.career_code, m1.age, m1.gender, m1.branch_no, m1.income_trade_status,\n" +
                "       m1.buyback_trade_status, m1.income_active_falg, m1.buyback_active_falg,\n" +
                "       m1.is_prod_cash, m1.is_qrpagent,m1.emp_name, m2.*, m3.profit_rate_y_label, m3.loss_rate_y_label, m3.buy_product_risk_label,\n" +
                "       m5.all_entrst_num, m5.all_entrst_amt, m6.no_trd_days, m6.no_login_days, m6.buy_habit_label, m6.activity_label,\n" +
                "       m6.turnover_label, m6.marg_sec_perfer_label, m6.marg_ratation_label,\n" +
                "       m6.marg_debit_change_label, m6.total_cms_y, m6.total_match_amt_y, m6.asset_outflow_rate, m6.turnover_y,\n" +
                "       m6.activity_y, m6.marg_ratation_m\n" +
                "from (\n" +
                "  select client_id, cust_type, normal_status, marg_status, cust_channel, risk_level,\n" +
                "         edu_code, career_code, age, gender, branch_no, income_trade_status,\n" +
                "         buyback_trade_status, income_active_falg, buyback_active_falg,\n" +
                "         is_prod_cash, is_qrpagent,emp_name from dws_cust_portrait_cust_base_info\n" +
                ") as m1\n" +
                "left join (\n" +
                "    select client_id, total_ast,net_ast, nor_total_ast, marg_net_ast, cap_bal, nor_cap_bal, marg_cap_bal,\n" +
                "           total_ast_peak, marg_net_ast_peak, mfin_val, stock_position_rate_last_y, avg_marg_cap_bal_m,\n" +
                "           avg_marg_cap_bal_y, avg_marg_net_ast_m, avg_nor_cap_bal_m, avg_nor_cap_bal_y, avg_total_ast_y,\n" +
                "           nor_ast_branch_level, nor_ast_company_level, marg_ast_branch_level ,marg_ast_company_level, total_ast_branch_level, total_ast_company_level, \n" +
                "           total_ast_label, marg_net_ast_label, marg_cap_bal_label, stock_position_label, stock_hold_label, sec_mkt_val\n" +
                "    from dws_cust_portrait_ast_info\n" +
                ") as m2\n" +
                "on m1.client_id = m2.client_id\n" +
                "left join (\n" +
                "    select client_id, profit_rate_y_label, loss_rate_y_label, buy_product_risk_label\n" +
                "    from dws_cust_portrait_cust_behavior_ablity_info\n" +
                ") as m3\n" +
                "on m1.client_id = m3.client_id\n" +
                "left join (\n" +
                "    select client_id, all_entrst_num, all_entrst_amt from dws_cust_portrait_cust_entrust_terminal\n" +
                ") as m5\n" +
                "on m1.client_id = m5.client_id\n" +
                "left join (\n" +
                "    select client_id, no_trd_days, no_login_days, buy_habit_label, activity_label,\n" +
                "           turnover_label, marg_sec_perfer_label, marg_ratation_label,\n" +
                "           marg_debit_change_label, total_cms_y, total_match_amt_y, asset_outflow_rate, turnover_y,\n" +
                "           activity_y, marg_ratation_m from dws_cust_portrait_cust_trd_info\n" +
                ") as m6\n" +
                "on m1.client_id = m6.client_id");
        // 写入HDFS文件
        sqlDF.coalesce(1).write()
                .format("csv")
                .option("header", "true")
                .option("mode", "overwrite")
                .save(fileName);
    }
}
